package com.shujia.flink.source

import org.apache.flink.streaming.api.functions.source.SourceFunction
import org.apache.flink.streaming.api.scala._

object Demo4MySOurce {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    //使用自定义source,
    //flink启动之后会取调用mySource对象的run获取数据
    val myDS: DataStream[Int] = env.addSource(new MySource)

    myDS.print()

    env.execute()


  }

}

/**
  * 自定义source
  *
  */
class MySource extends SourceFunction[Int] {
  /**
    * 生产获取数据，比如可以使用jdbc从mysql中读取数据
    *
    * run  方法只会调用一次
    *
    * @param ctx 用于将数据发送到下游
    */
  override def run(ctx: SourceFunction.SourceContext[Int]): Unit = {

    var i = 0
    while (true) {

      //将数据发送到下游
      ctx.collect(i)

      Thread.sleep(1000)

      i += 1
    }
  }

  //回收资源
  override def cancel(): Unit = {

  }
}
